package spring.cloud.tasks.client.item_task_processor.task_items_processor;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

@Slf4j
public class WaitWhenNoDataModeTaskProcessingThread extends Thread {
    public WaitWhenNoDataModeTaskProcessingThread(WaitWhenNoDataModeTaskItemsProcessor taskProcessor) {
        this.taskProcessor = taskProcessor;
    }

    private WaitWhenNoDataModeTaskItemsProcessor taskProcessor;

    public void run() {
        try {
            long startTime = 0;
            while (true) {
                taskProcessor.signal.increaseThreadCount1();
                Object taskDataId;
                while (true) {
                    //destroy
                    if (taskProcessor.destroy) {
                        taskProcessor.signal.decreaseThreadCount1();
                        taskProcessor.signal.notifyOtherThreadsBeAwakened();
                        //
                        synchronized (taskProcessor.threadList) {
                            taskProcessor.threadList.remove(Thread.currentThread());
                        }
                        return;
                    }
                    //load task from here
                    if (taskProcessor.multipleTaskDataProcessor == false) {
                        taskDataId = taskProcessor.getTaskDataId();
                    } else {
                        taskDataId = taskProcessor.getTaskDataIds();
                    }

                    if (taskDataId == null) {
                        //跳出内层循环
                        break;
                    }
                    try {
                        startTime = taskProcessor.ruleTaskProcessor.zooKeeperDataManager.zooKeeperCurrentTimeMillis();
                        if (!taskProcessor.multipleTaskDataProcessor)//SingleTaskDataProcessor
                        {
                            if (taskProcessor.taskDataProcessor.execute(taskDataId) == true) {
                                taskProcessor.addTaskSuccessNum(1, taskProcessor.ruleTaskProcessor.zooKeeperDataManager.zooKeeperCurrentTimeMillis() - startTime);
                            } else {
                                taskProcessor.addTaskFailNum(1, taskProcessor.ruleTaskProcessor.zooKeeperDataManager.zooKeeperCurrentTimeMillis() - startTime);
                            }
                        } else {//MultipleTaskDataProcessor
                            if (taskProcessor.taskDataProcessor.execute((Object[]) taskDataId) == true) {
                                taskProcessor.addTaskSuccessNum(((Object[]) taskDataId).length, taskProcessor.ruleTaskProcessor.zooKeeperDataManager.zooKeeperCurrentTimeMillis() - startTime);
                            } else {
                                taskProcessor.addTaskFailNum(((Object[]) taskDataId).length, taskProcessor.ruleTaskProcessor.zooKeeperDataManager.zooKeeperCurrentTimeMillis() - startTime);
                            }
                        }
                    } catch (Throwable ex) {
                        if (!taskProcessor.multipleTaskDataProcessor) {
                            taskProcessor.addTaskFailNum(1, taskProcessor.ruleTaskProcessor.zooKeeperDataManager.zooKeeperCurrentTimeMillis() - startTime);
                        } else {
                            taskProcessor.addTaskFailNum(((Object[]) taskDataId).length, taskProcessor.ruleTaskProcessor.zooKeeperDataManager.zooKeeperCurrentTimeMillis() - startTime);
                        }
                        log.warn("TaskDataId :" + taskDataId + " ", ex);
                    }
                }//inner layer while

                if (taskProcessor.signal.decreaseThreadCount1IfThreadCountNotEqual1AndReturnWhetherThreadCountIsNot1()) {
                    /**
                     * 没有数据且不是最后一个线程需要等被唤醒【核心要点】
                     */
                    taskProcessor.signal.waitBeAwakened();
                } else {
                    //只有这最后一个线程有对threadCount减去1的操作
                    TimeUnit.SECONDS.sleep(1);
                    startTime = taskProcessor.ruleTaskProcessor.zooKeeperDataManager.zooKeeperCurrentTimeMillis();
                    int taskDataSize = taskProcessor.getTaskDataIdsFromDb();
                    if (taskDataSize > 0) {
                        taskProcessor.signal.notifyOtherThreadsBeAwakened();//通知其他线程进行处理
                    } else {
                        if (!taskProcessor.destroy && taskProcessor.ruleTaskProcessor.isContinueWhenNoTaskDataIdList()) {
                            taskProcessor.isSleeping = true;
                            Thread.sleep(taskProcessor.ruleTaskProcessor.getTask().getSleepTimeWhenNoData());
                            taskProcessor.isSleeping = false;
                        } else {
                            taskProcessor.signal.notifyOtherThreadsBeAwakened();
                        }
                    }
                    taskProcessor.signal.decreaseThreadCount1();
                }
            }// outer layer while
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
        }
    }
}
